package Kafka_model

import java.sql.ResultSet

import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.log4j.{Level, Logger}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe

object mykafka {
	def main(args: Array[String]): Unit = {
		println("环境搭建成功")
		Logger.getLogger("org").setLevel(Level.WARN)
		System.setProperty("hadoop.home.dir","D:\\work\\hadoop-2.9.2")
		val conf=new SparkConf().setMaster("local[3]").setAppName("kafka")
		val ssc=new StreamingContext(conf,Seconds(5))
		ssc.checkpoint("d:\\cj\\streaming")

		val kafkaParams=Map[String,Object](
			"bootstrap.servers"->"192.168.18.50:9092",
			"key.deserializer"->classOf[StringDeserializer],
			"value.deserializer"->classOf[StringDeserializer],
			"group.id"->"kafka01"
		)

//		println("环境搭建完成")
//		Logger.getLogger("org").setLevel(Level.WARN)
//
//		val conf = new SparkConf().setMaster("local[3]").setAppName("kafka")
//
//		val ssc = new StreamingContext(conf, Seconds(5))
//		ssc.checkpoint("D:\\项目\\streaming\\game")
//
//		val kafkaParams = Map[String, Object](
//			"bootstrap.servers" -> "192.168.215.10:9092",
//			"key.deserializer" -> classOf[StringDeserializer],
//			"value.deserializer" -> classOf[StringDeserializer],
//			"group.id" -> "kafka01"
//		)

		val topics=Array("game_list")
		println("==================start==================")
		var rs:ResultSet=null
		println("===================end===================")


		val kafkaStream=KafkaUtils.createDirectStream[String,String](ssc,PreferConsistent,Subscribe[String,String](topics,kafkaParams))

		val ds=kafkaStream
  		.map(school => (school.value()))
  		.foreachRDD((rdd,index)=>{
			println("--------------------"+index)
			rdd.foreach(item=>{
				val data = JSON.parseFull(item).get.asInstanceOf[Map[String, Any]]
				val gameData = new Games(data)
				val SELECT = s"SELECT gname FROM game WHERE gname = '${gameData.gname}'"

				println(s"SELECT:${SELECT}")
				println(s"description:${gameData.description}")

				rs = myConn.conn().createStatement().executeQuery(SELECT)
				if (!rs.next()) {
					//数据库没有相同数据
					println("not have")
					val sql = s"INSERT INTO game (" +
						s"gname," +
						s"author," +
						s"score," +
						s"description," +
						s"category," +
						s"grank" +
						s")"+
						s" VALUES (" +
						s"'${gameData.gname}','${gameData.author}'," +
						s"'${gameData.score}','${gameData.description}'" +
						s",'${gameData.category}','${gameData.grank}'"+
						")"
					println(s"insert:${sql}")
					myConn.conn().createStatement().executeUpdate(sql)
				} else {
					//数据库有相同数据
					println("has")
					val sql = s"UPDATE game" +
						s" SET" +
						s" author = '${gameData.author}'," +
						s" score = '${gameData.score}'," +
						s" description = '${gameData.description}'," +
						s" category = '${gameData.category}'," +
						s" grank = '${gameData.grank}'" +
						s"WHERE gname = '${gameData.gname}'"
					println(s"updateSql:${sql}")
					myConn.conn().createStatement().executeUpdate(sql)
				}
			})
		})
		print(ds)
		ssc.start()
		ssc.awaitTermination()
	}
}
